package com.atguigu.edu.realtime.app.dwd.course;

import com.atguigu.edu.realtime.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @date 2022-08-31
 * @author maikle
 * @explain dwd层一个用户对一个课程的评分
 */
public class DwdCourseReview {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(3);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 设置状态的失效时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15*60 + 10));

        //TODO 2.检查点相关的设置(略)

        //TODO 3.从kafka的topic_db主题中读取数据 创建对应的动态表
        tableEnv.executeSql("create table ods_db(" +
                "`database` String,\n" +
                "`table` String,\n" +
                "`type` String,\n" +
                "`data` map<String, String>,\n" +
                "`old` map<String, String>,\n" +
                "`proc_time` as PROCTIME(),\n" +
                "`ts` string\n" +
                ")" + KafkaUtil.getKafkaDDL("ods_db", "dwd_trade_order_pre_process"));



        // todo: 4. 从ods_db表中过滤出课程评价信息
        Table review_info = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['user_id'],\n" +
                "data['course_id'],\n" +
                "data['review_stars'],\n" +

                "data['create_time'],\n" +
                "date_format(data['create_time'], 'yyyy-MM-dd') create_date,\n" +
                "current_row_timestamp() row_op_ts\n" +
                "from ods_db\n" +
                "where `table` = 'review_info'\n" +
                "and `type` = 'insert'\n" +
                "and data['deleted'] = '0'"
        );
        tableEnv.createTemporaryView("review_info", review_info);
        //tableEnv.executeSql("select * from review_info").print();
        // todo: 5. 创建课程评价信息表和要写出kafka主题进行映射
        tableEnv.executeSql("" +
                "create table dwd_course_review(\n" +
                "id string,\n" +
                "user_id string,\n" +
                "course_id string,\n" +
                "review_stars string,\n" +
                "create_time string,\n" +
                "create_date string,\n" +
                "row_op_ts timestamp_ltz(3),\n" +
                "primary key(id) not enforced\n" +
                ")" + KafkaUtil.getUpsertKafkaDDL("dwd_course_review"));

        // todo: 6. 将数据写到kafka对应的主题中
        tableEnv.executeSql("insert into dwd_course_review \n" +
                "select * from review_info");
    }
}
